package UserMapReduce;

import java.io.BufferedReader;
import java.io.DataInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.sql.Driver;
import java.util.ArrayList;
import java.util.StringTokenizer;
import java.util.Vector;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import dataStructures.CodeBook;
import dataStructures.ProfileWritable;
import dataStructures.Ratings;
import dataStructures.UsersMapOutputWritable;


public class UserProcessMapper extends MapReduceBase implements Mapper<NullWritable, /*Text*/ ProfileWritable , IntWritable, UsersMapOutputWritable>{
// export codeBookItem
	private int K; // ITEM CLUSTER
	private int L; //USER CLUSTER
	private File codebookUser;
	private Vector<CodeBook> cb;
	
	@Override
	 public void configure(JobConf job) {
         K = job.getInt("K" , -1);
         	if(K == -1 )
         		throw new RuntimeException("Cant read K");
         
         	L = job.getInt("L" , -1);
         	if(L == -1 )
         		throw new RuntimeException("Cant read L"); 	
         	
         	
        codebookUser = new File (job.get("CodeBookUserPath"));
        cb = CodeBook.fromFile(codebookUser);
         	if(codebookUser == null )
         		throw new RuntimeException("Cant read codeBookUser");
         
       }
	
	@Override
	public void map(NullWritable key, /*Text*/ ProfileWritable val,
			OutputCollector<IntWritable, UsersMapOutputWritable > output, Reporter arg3)
			throws IOException {
			
			//ProfileWritable val = ProfileWritable.fromString(val1.toString());
			int [] itemID = {val.getRater()};
			int clusterID = getOptimizedCluster(val);
			UsersMapOutputWritable u  = new UsersMapOutputWritable(itemID , calcRatingSums(val.getRated(), val.getRating())) ;
			output.collect(new IntWritable(clusterID), u);
	
		
	}
	
	private int getOptimizedCluster(ProfileWritable val){
		int[] items = val.getRated();
		int[] ratings = val.getRating();
		int userId = val.getRater();
		int minTotalError = Integer.MAX_VALUE;
		int bestCluster = 0;
		int sum = 0;
		
		for (int indexClusterUser=0; indexClusterUser<this.L; indexClusterUser++){//iterate all user possible Clusters
			sum = 0;
			for (int j=0; j<items.length; j++){
				//find the cluster of item i in items
				int clusterNumOfItem = getClusterOfItem(items[j], cb);
				//find and add get rating of item in cluster that found
				double arrOfValues[] = getArrayOfIds(clusterNumOfItem, cb);
				double ratingInCb = arrOfValues[indexClusterUser];
				//discount ratings of items i from rating in item in cluster that found
				sum += Math.pow((ratings[j] - ratingInCb), 2);
			}
			if (sum < minTotalError){
				minTotalError = sum;
				bestCluster = indexClusterUser;
			}
		}
		return bestCluster;
	}
	
	private int getClusterOfItem (int itemId, Vector<CodeBook> cb){
		for (int i=0; i<cb.size(); i++){
			for (int j=0; j<cb.get(i).getUserIds().length; j++){
				if (cb.get(i).getUserIds()[j] == itemId){
					return cb.get(i).getClusterId();
				}
			}
		}
		System.out.println(itemId + ":No Cluster Found in User Mapper");
		return 0; //default cluster
	}
	
	private double[] getArrayOfIds(int clusterId, Vector<CodeBook> cb){
		for (int i=0; i<cb.size(); i++){
			if (cb.get(i).getClusterId() == clusterId)
				return cb.get(i).getCodeValues();
		}
		System.out.println("No Ids of cluster was found in User Mapper");
		return new double[1];
		
	}


	private Ratings [] calcRatingSums(int[] rated  , int[] rating) {
		//Build the cb elemnts for the users
		Ratings [] array = new Ratings[K];//
		for(int cluster = 0 ; cluster < K ; cluster++){
			array[cluster] = buildCBElementToCluster(rated , cluster , rating);
		}
		
		return array;
		
	}

	private Ratings buildCBElementToCluster(int [] IDs , int clusterID , int [] grades) {
		int cluster;
		StringTokenizer st2;
		String line;
		String arrayOfIDsMembers;
		try{
			  FileInputStream fstream = new FileInputStream(codebookUser);
			  // Get the object of DataInputStream
			  DataInputStream in = new DataInputStream(fstream);
			  BufferedReader br = new BufferedReader(new InputStreamReader(in));
			  //Read File Line By Line
			  while ((line = br.readLine()) != null)   {
				  st2 = new StringTokenizer(line , "\t\n");
					cluster = Integer.parseInt(st2.nextToken());
					if(cluster == clusterID){ //the cluster was found
						arrayOfIDsMembers = st2.nextToken(); //members in the cluster
						in.close();
						return buildCBElement(IDs , arrayOfIDsMembers , grades);
					}
			  
			  }
			  
			  return new Ratings(0,0);
			  //throw new RuntimeException("The cluster" + clusterID +" was not found in the codebook");
			  
			    }catch (Exception e){//Catch exception if any
			    	e.printStackTrace();
			    	System.err.println("Error: " + e.getMessage());
			    	System.exit(-1);
			  }
			  
	return null;
	
	}

	private Ratings buildCBElement(int[] iDs, String arrayOfIDsMembers , int[] grades) {
		ArrayList<Integer> a = new ArrayList<Integer>(); //the array of ids that belongs to the cluster
		StringTokenizer st = new StringTokenizer(arrayOfIDsMembers , "[]\t\n, ");
		int counter = 0;
		int sum = 0;
		while(st.hasMoreTokens()){
			a.add(Integer.parseInt(st.nextToken()));
		}
		
		for(int i = 0 ; i < iDs.length ; i++){
			if( a.contains(new Integer(iDs[i]))){
				counter++;
				sum += grades[i];
			}
		}
		
		return new Ratings(counter , sum);
	}
}
